package com.shujia.spark.core



import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo12Action {
  def main(args: Array[String]): Unit = {
    //配置spark
    val conf: SparkConf = new SparkConf().setAppName("action").setMaster("local")

    //创建spark对象
    val sc = new SparkContext(conf)

    //读取学生表数据
    val students: RDD[String] = sc.textFile("data/students.txt")

    /**
      * action算子：起到触发任务执行的作用，每一个action都会触发一个job
      */

    /**
      * count:统计rdd的数量
      */
    val count: Long = students.count()
    println(count)

    /**
      * collect:将rdd的数据拉取到内存中，变成一个数组
      *
      * 如果rdd的数据量超过了内存会出现内存溢出
      */
    val array: Array[String] = students.collect()

    /**
      * reduce:全局聚合
      */

    val reduce: String = students.reduce(_ + _)

    /**
      * foreach：遍历rdd中的数据
      *
      * foreachPartition：
      * 遍历一个分区中的数据，一般用于将数据保留在外部数据库的时候
      */
   //students.foreach(println)
    students.foreachPartition((iter: Iterator[String])=>{
      iter.foreach(println)
    })

    /**
      * saveAsTextFile保存数据,如果输出目录已存在会报错
      */
    //创建HDFS配置文件对象
    val configuration = new Configuration()

    //创建fileSystem对象
    val fileSystem: FileSystem = FileSystem.get(configuration)

    //判断输出目录是否存在
    val path = new fs.Path("data/out1")
    if (fileSystem.exists(path)){
      //如果目录存在删除目录
      fileSystem.delete(path,true)
    }
    students.saveAsTextFile("data/out1")
  }
}
